package com.smsc.headend.task.engine.utils;

import cn.hutool.core.date.SystemClock;
import com.smsc.headend.module.data.entity.RdMeterReadsLp;
import com.smsc.headend.module.task.dto.TaskReadData;
import com.smsc.headend.task.engine.service.asset.AssetService;
import com.smsc.headend.task.engine.service.dds.DdsService;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;
import org.springframework.util.CollectionUtils;

import javax.annotation.PostConstruct;
import java.util.*;
import java.util.concurrent.*;
import java.util.stream.Collectors;


@Component
@Slf4j
public class DdsBatchGenerateUtil {

    ScheduledExecutorService persistentService;

    BlockingQueue<TaskReadData> ddsReadDataQueue = new LinkedBlockingQueue();

    BlockingQueue<RdMeterReadsLp> ddsQueue = new LinkedBlockingQueue();

    @Autowired
    AssetService assetService;

    @Autowired
    DdsService ddsService;

    @PostConstruct
    public void initSchedule() {
        persistentService = Executors.newSingleThreadScheduledExecutor();
        persistentService.scheduleAtFixedRate(() -> {
            Long start = SystemClock.now();
            Integer size = 0;
            try {
                List<TaskReadData> readDatas = new ArrayList<>();
                while (true) {
                    TaskReadData readData = ddsReadDataQueue.poll(10, TimeUnit.SECONDS);
                    if (readData == null) {
                        break;
                    }
                    readDatas.add(readData);
//                    String ddsFileBatchSize = assetService.getSystemProperty("DDS_FILE_BATCH_SIZE");
                    Integer batchThreshold = 50;
                    if (readDatas.size() >= batchThreshold) {
                        break;
                    }
                }
                if (CollectionUtils.isEmpty(readDatas)) {
                    return;
                }
                size = readDatas.size();
                List<RdMeterReadsLp> rdMeterReadsLps = readDatas.stream().flatMap(trd -> trd.getTaskMeterReads().stream()).collect(Collectors.toList());
                ddsService.transformMeterReadResult(rdMeterReadsLps);
                readDatas = null;
                log.debug("save dds cost: {}ms, dataSize={} ,left={} ", SystemClock.now() - start, size, ddsQueue.size());
            } catch (Exception e) {
                log.error("dds save failed", e);
            }

//            for (Integer integer : cachedReadRecordPoolMap.keySet()) {
//                log.info("pool {} size {}  counter {}", integer,  cachedReadRecordPoolMap.get(integer).getContainer().size(),  cachedReadRecordPoolMap.get(integer).getCounter().get());
//            }
        }, 0, 200, TimeUnit.MILLISECONDS);
    }

    public void insertTaskRecord(List<TaskReadData> taskReadDataList) {
        for (TaskReadData record : taskReadDataList) {
            Long meterId = CollectionUtils.isEmpty(record.getTaskMeterReads()) ? -1 : record.getTaskMeterReads().get(0).getMeterId();
            Long tv = CollectionUtils.isEmpty(record.getTaskMeterReads()) ? -1 : record.getTaskMeterReads().get(0).getTv();
            Long lpDataItemId = CollectionUtils.isEmpty(record.getTaskMeterReads()) ? -1 : record.getTaskMeterReads().get(0).getLpDataItemId();
            if (lpDataItemId.equals(1020070010L)) {
                log.debug("Dds: enterQueue meterId={}, lpDataItemId={}, tv={}", meterId, lpDataItemId, tv);
            } else {
                log.info("Dds: enterQueue meterId={}, lpDataItemId={}, tv={}", meterId, lpDataItemId, tv);
            }
            try {
                if(!ddsReadDataQueue.offer(record)) {
                    log.error("no data");
                }
            } catch (Exception e) {
                log.error("insert error", record, e);
            }
        }
    }


}
